package cn._51doit.day07;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @create: 2021-10-23 20:46
 * @author: 今晚打脑斧先森
 * @program: MapStateDemo
 * @Description:
 *    KeyedState是Flink中KeyedStream使用的跟key绑定的状态
 *    KeyedState有3种：
 *
 *    valueState
 *    	Map<key, value>
 *
 *    mapState
 *    	Map<key, Map<k, v>>
 *
 *    listState
 *    	Map<key, List<V>>
 *
 *
 *    需求：
 *    河北省,保定市,3000
 *    河北省,廊坊市,2000
 *    河北省,保定市,2000
 *
 *    辽宁省,沈阳市,1000
 *    辽宁省,大连市,2000
 *    辽宁省,大连市,2000
 *
 *    相同省份的数据分到同一个分区，并且按照城市统计金额
 *    按照省份进行keyBy
 **/
public class MapStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);

        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple3<String, String, Double>> map1 = lines.map(new MapFunction<String, Tuple3<String, String, Double>>() {
            @Override
            public Tuple3<String, String, Double> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], Double.parseDouble(split[2]));
            }
        });
        KeyedStream<Tuple3<String, String, Double>, String> keyedStream = map1.keyBy(t -> t.f0);
        SingleOutputStreamOperator<Tuple4<String, Double, String, Double>> res = keyedStream.process(new ProcessFunction<Tuple3<String, String, Double>, Tuple4<String, Double, String, Double>>() {

            //使用transient修饰,这样客户端就没法更改它了,因为它就不参与反序列化和序列化了
            private transient MapState<String, Double> mapState;
            @Override
            public void open(Configuration parameters) throws Exception {
                //定义描述器
                MapStateDescriptor<String, Double> mapStateDescriptor = new MapStateDescriptor<>("省份城市,金额", String.class, Double.class);
                //根据描述器 初始化和恢复状态
                mapState = getRuntimeContext().getMapState(mapStateDescriptor);
            }

            @Override
            public void processElement(Tuple3<String, String, Double> value, Context ctx, Collector<Tuple4<String, Double, String, Double>> out) throws Exception {
                String city = value.f1;
                Double money = value.f2;

                //获取历史数据
                Double historyMoney = mapState.get(city);
                if (historyMoney == null) {
                    historyMoney = 0.0;
                }
                historyMoney += money;
                //更新状态,因为mapState没有update方法,所以put就行了
                mapState.put(city, historyMoney);

                //总的金额
                Double totalCountMoney = 0.0;
                Iterable<Double> doubleIterable = mapState.values(); //获得所有的Double数据
                for (Double aDouble : doubleIterable) {
                    totalCountMoney += aDouble;
                }
                out.collect(Tuple4.of(value.f0, historyMoney, city, totalCountMoney));
            }
        });
        res.print();
        env.execute();

    }
}
